package com.niodata.dp.core.hdfs.legacy;

import static org.apache.hadoop.fs.CommonConfigurationKeysPublic.HADOOP_SECURITY_AUTHENTICATION;

import com.niodata.dp.core.ConfigurationKeys;
import com.niodata.dp.core.DpConfiguration;
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.net.URI;
import java.net.URISyntaxException;
import java.nio.file.Paths;
import java.security.PrivilegedAction;
import java.security.PrivilegedExceptionAction;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.permission.FsPermission;
import org.apache.hadoop.security.UserGroupInformation;
import org.apache.hadoop.security.UserGroupInformation.AuthenticationMethod;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * raw file system over hdfs.
 *
 * @author huqiao
 */
public class RawHdfsFileSystem {

  private static FileSystem hdfs = null;
  private static UserGroupInformation ugi;

  static {
    System.setProperty("hadoop.home.dir",
          DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME));
    try {
      String proxyUser = DpConfiguration.getString(ConfigurationKeys.HDFS_PROXY_USER);
      System.setProperty("java.security.krb5.conf",
            DpConfiguration.getString(ConfigurationKeys.KERBEROS_CONF, ""));
      String proxyUserKeyTab =
            DpConfiguration.getString(ConfigurationKeys.HDFS_PROXY_USER_KEYTAB, "");
      login(proxyUser, proxyUserKeyTab);
    } catch (IOException e) {
      throw new RuntimeException(e);
    }
  }

  private final Logger logger = LoggerFactory.getLogger(getClass());
  private final String realuser;

  private RawHdfsFileSystem(String realuser) {
    this.realuser = realuser;
  }


  public static RawHdfsFileSystem getInstance(String realuser) throws IOException {
    RawHdfsFileSystem fs = new RawHdfsFileSystem(realuser);
    return fs;
  }

  private static void login(String user, String keytab) throws IOException {

    Configuration ugiConf = new Configuration();
    if (DpConfiguration.isHadoopKerberosEnabled()) {
      ugiConf.set(HADOOP_SECURITY_AUTHENTICATION, AuthenticationMethod.KERBEROS.name());
      UserGroupInformation.setConfiguration(ugiConf);
      ugi = UserGroupInformation.loginUserFromKeytabAndReturnUGI(user, keytab);
    } else {
      ugi = UserGroupInformation.createRemoteUser(user);
    }

    try {
      ugi.doAs(
            new PrivilegedExceptionAction<Void>() {

              @Override
              public Void run() throws Exception {
                Configuration conf = new Configuration();
                conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
                String coreSite =
                      DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME)
                            + "/etc/hadoop/core-site.xml";
                String hdfsSite =
                      DpConfiguration.getString(ConfigurationKeys.HADOOP_HOME)
                            + "/etc/hadoop/hdfs-site.xml";
                conf.addResource(new Path(coreSite));
                conf.addResource(new Path(hdfsSite));
                URI uri;
                try {
                  uri = new URI(DpConfiguration.getString(ConfigurationKeys.HDFS_URI));
                } catch (URISyntaxException ex) {
                  throw new RuntimeException(ex);
                }
                hdfs = FileSystem.get(uri, conf);
                return null;
              }
            });
    } catch (InterruptedException ex) {
      throw new IOException(ex);
    }
  }

  public FileStatus getFileStatus(String path) throws IOException {
    FileStatus status =
          doPrivileged(
                new PrivilegedExceptionAction<FileStatus>() {
                  @Override
                  public FileStatus run() throws Exception {
                    return hdfs.getFileStatus(new Path(path));
                  }
                });
    return status;
  }

  public FileStatus[] listStatus(String dir) throws IOException {
    FileStatus[] hdfsFiles =
          doPrivileged(
                new PrivilegedExceptionAction<FileStatus[]>() {
                  @Override
                  public FileStatus[] run() throws Exception {
                    return hdfs.listStatus(new Path(dir));
                  }
                });
    return hdfsFiles;
  }

  private <T> T doPrivileged(PrivilegedExceptionAction<T> action) throws IOException {
    ugi.checkTGTAndReloginFromKeytab();
    try {
      return UserGroupInformation.createProxyUser(realuser, ugi).doAs(action);
    } catch (InterruptedException e) {
      throw new RuntimeException(e);
    }
  }

  private <T> T doPrivileged(PrivilegedAction<T> action) throws IOException {
    ugi.checkTGTAndReloginFromKeytab();
    return UserGroupInformation.createProxyUser(realuser, ugi).doAs(action);
  }

  /**
   * make directory.
   *
   * @param path path
   * @param permission dir permission
   * @return true iff mkdir successfully
   * @throws IOException error
   */
  public boolean mkdir(String path, FsPermission permission) throws IOException {

    return doPrivileged(
          new PrivilegedExceptionAction<Boolean>() {

            @Override
            public Boolean run() throws IOException {
              String parent = Paths.get(path).getParent().toString();
              if (!hdfs.exists(new Path(parent))) {
                throw new IOException("parent directory not exist");
              }
              if (hdfs.isFile(new Path(parent))) {
                throw new IOException("parent directory exists as a file.");
              }
              boolean result = hdfs.mkdirs(new Path(path), permission);
              if (result) {
                hdfs.setOwner(new Path(path), realuser, null);
              }
              return result;
            }
          });
  }


  /**
   * delete file.
   *
   * @param path path
   * @param recursive rec
   * @throws IOException ex
   */
  public void delete(String path, boolean recursive) throws IOException {
    doPrivileged(
          new PrivilegedExceptionAction<Boolean>() {

            @Override
            public Boolean run() throws Exception {
              hdfs.delete(new Path(path), recursive);
              return null;
            }
          }
    );
  }

  public void copyFromLocal(String src, String target, FsPermission permission) throws IOException {
    doPrivileged(
          new PrivilegedExceptionAction<Void>() {
            @Override
            public Void run() throws Exception {
              Path srcPath = new Path(src);
              hdfs.copyFromLocalFile(srcPath, new Path(target));
              hdfs.setOwner(new Path(target), realuser, null);
              hdfs.setPermission(new Path(target), permission);
              return null;
            }
          });
  }

  public void copyToLocal(String src, String target) throws IOException {
    doPrivileged(
          new PrivilegedExceptionAction<Void>() {
            @Override
            public Void run() throws Exception {
              Path path = new Path(src);
              hdfs.copyToLocalFile(path, new Path(target));
              return null;
            }
          });
  }

  public boolean isDirectory(String path) throws IOException {
    return hdfs.isDirectory(new Path(path));
  }

  public long getCapacity() throws IOException {
    return hdfs.getStatus().getCapacity();
  }

  public long getUsed() throws IOException {
    return hdfs.getUsed();
  }

  public boolean isFile(String path) throws IOException {
    return hdfs.isFile(new Path(path));
  }

  public long getFileSize(String path) throws IOException {
    FileStatus fileStatus = hdfs.getFileStatus(new Path(path));
    return fileStatus.getLen();
  }

  /**
   * write content to file.
   */
  public long write(String path, InputStream os, boolean overwrite, FsPermission permission)
        throws IOException {
    return doPrivileged(
          new PrivilegedExceptionAction<Long>() {
            @Override
            public Long run() throws Exception {
              long size = 0;
              logger.info("start writing to hdfs:{}", path);
              FSDataOutputStream out = hdfs.create(new Path(path), overwrite);
              try {
                size = IOUtils.copyLarge(os, out);
                out.flush();
              } catch (IOException ex) {
                throw ex;
              } finally {
                out.close();
              }
              logger.info("write to {} successfully.", path);
              logger.debug("set owner of {} to {}", path, realuser);
              hdfs.setOwner(new Path(path), realuser, null);
              logger.debug("set permission for {}", path);
              hdfs.setPermission(new Path(path), permission);
              return size;
            }
          });
  }

  public InputStream open(String path) throws IOException {
    return doPrivileged(
          new PrivilegedExceptionAction<InputStream>() {

            @Override
            public InputStream run() throws Exception {
              return hdfs.open(new Path(path));
            }
          });
  }

  public OutputStream create(String path, boolean overwrite) throws IOException {
    return doPrivileged(
          new PrivilegedExceptionAction<OutputStream>() {

            @Override
            public OutputStream run() throws Exception {
              return hdfs.create(new Path(path), overwrite);
            }
          });
  }


  public void close() throws IOException {
    ugi.logoutUserFromKeytab();
    hdfs.close();
  }

  public boolean exists(String path) throws IOException {
    return doPrivileged(new PrivilegedExceptionAction<Boolean>() {
      @Override
      public Boolean run() throws IOException {
        return hdfs.exists(new Path(path));
      }
    });
  }

  /**
   * 创建目录.
   *
   * @param path path
   * @param permission permission
   * @throws IOException ex
   */
  public boolean mkdirs(String path, FsPermission permission) throws IOException {
    return doPrivileged(new PrivilegedExceptionAction<Boolean>() {
      @Override
      public Boolean run() throws IOException {
        Path pa = new Path(path);
        if (pa.getParent() == null) {
          return false;
        }
        if (hdfs.exists(pa.getParent())) {
          return mkdir(path, permission);
        } else {
          return mkdirs(new Path(path).getParent().toString(), permission) && mkdir(path,
                permission);
        }
      }
    });
  }


  public boolean mkdirs(String path) throws IOException {
    return doPrivileged(new PrivilegedExceptionAction<Boolean>() {
      @Override
      public Boolean run() throws IOException {
        Path pa = new Path(path);
        if (pa.getParent() == null) {
          return false;
        }
        return hdfs.mkdirs(new Path(path));
      }
    });
  }

  public boolean rename(Path src, Path dist) throws IOException {
    return doPrivileged(new PrivilegedExceptionAction<Boolean>() {
      @Override
      public Boolean run() throws IOException {
        return hdfs.rename(src, dist);
      }
    });
  }
}